package ds_industry_2025.ds.ds_06.T5




import org.apache.spark.mllib.linalg.{DenseVector, Matrix, SingularValueDecomposition, Vector, Vectors}
import org.apache.spark.mllib.linalg.distributed.RowMatrix
import org.apache.spark.mllib.linalg.distributed.RowMatrix
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.functions._

import scala.collection.mutable.ArrayBuffer

/*
    1、根据子任务一的结果，对其进行SVD分解，对数据进行降维保留前5个奇异值信息，根据该用户已购买的商品分别与未购买的商品计算余弦相
    似度再进行累加求均值，将均值最大的5件商品id进行输出作为推荐使用。将输出结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】
    中对应的任务序号下。
结果格式如下：

------------------------推荐Top5结果如下------------------------
相似度top1(商品id：1，平均相似度：0.983456)
相似度top2(商品id：71，平均相似度：0.782672)
相似度top3(商品id：22，平均相似度：0.7635246)
相似度top4(商品id：351，平均相似度：0.7335748)
相似度top5(商品id：14，平均相似度：0.522356)

 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("推荐系统")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    //  todo 导入隐式转换
    import spark.implicits._

    // todo 读取特征工程生成的答案
    val data = spark.table("tzgc.skumat")

    //  todo 过滤出用户0.0购买和为购买的商品的数据情况
    //  todo 这里选择1是因为题目要求是把特征工程里面第二题的结果拿来做这一题，所以只需要提取第一个用户的购买信息
    val seq:Seq[Double] = data.filter(col("user_id") === 0.0).limit(1)
      .map(
        _.toSeq.map(_.toString.toDouble)
      )
      .collect()(0)
    // 使用切片函数除了用户是0.0，其余的全部取出来
    val skus:Seq[Double] = seq.slice(1, seq.length)

    //  分离已购买和未购买的商品
    val buy=new ArrayBuffer[Int]()
    val nobuy=new ArrayBuffer[Int]()

    // zipWithIndex 是 Scala 中的一个方法，它将序列中的每个元素与其索引配对，生成一个新的序列，每个元素是
    // 一个元组 (value, index)。
    skus.zipWithIndex.foreach(
      e=> {
        //  这里的e._1就是取元组里面的第一个元素
        if(e._1.equals(0.0)){
          //  如果是0就是没有购买这个商品，则添加到noBuy数组里面
          nobuy +=e._2
        }else{
          //  其余的就是
          buy +=e._2
        }
      }
    )


    //  todo 将 data 中的每一行数据转换为一个稠密向量（DenseVector），并将所有向量收集到一个数组中。
    //  Vectors.dense:创建一个稠密向量
    /*
        RDD.map：

    1.返回的是一个 RDD[U]，其中 U 是映射后的类型。
    2.不需要编码器，可以直接对数据进行任意操作。
    3.适合处理非结构化数据或需要复杂操作的场景。
    4.在这段代码中，目标是将 DataFrame 的每一行转换为一个 Vector 对象。由于 Vector 是一个非结构化数据类型，使用 RDD 的 map 方法更为方便。
     */
    val array:Array[Vector] = data.rdd.map(
      (r: Row) => {
        val vector: Vector = Vectors.dense(r.toSeq.slice(1, r.toSeq.length).map((_: Any).toString.toDouble).toArray)
        vector
      }
    ).collect()

    /*
        RowMatrix(行矩阵) 的作用
    RowMatrix 是一个分布式矩阵类，适用于大规模数据集的线性代数计算。它的特点包括：

    每行是一个局部向量（org.apache.spark.mllib.linalg.Vector）。

    数据分布存储在集群的多个节点上。

    支持矩阵乘法、奇异值分解（SVD）、主成分分析（PCA）等操作。

    在Apache Spark中，RowMatrix是一个分布式矩阵类，它的数据必须存储在一个RDD[Vector]中，而不是普通的Scala数组（Array）。
    因此，直接传递一个Array给RowMatrix构造函数是不合法的，因为RowMatrix期望的是一个分布式的数据结构（即RDD），而不是本地的数
    组。spark.sparkContext.parallelize(array)的作用是将本地的Array转换为一个分布式的RDD。这样，Spark可以将数据分布到集群
    的多个节点上，并进行并行计算。
     */
    //  todo 将数据转换为 RowMatrix
    val rowMatrix:RowMatrix = new RowMatrix(spark.sparkContext.parallelize(array))
    // todo 进行svd分解，保留前 5个奇异值
    //  computeSVD：进行奇异值分解 里面的5表示保留前5个奇异值
    //  computeU = true：表示计算左奇异向量矩阵U。如果不需要U，可以设置为false以节省计算资源
    //  返回一个SingularValueDecomposition对象，包含三个部分：U（左奇异向量矩阵）、Σ（奇异值矩阵）、V（右奇异向量矩阵）。
    val svd:SingularValueDecomposition[RowMatrix,Matrix]=rowMatrix.computeSVD(5,computeU=true)
    //  todo svd.V：从SVD结果中提取右奇异向量矩阵V。V是一个Matrix对象，表示右奇异向量矩阵。
    val v:Matrix=svd.V



    //  todo rowIter 是矩阵 V 的行迭代器，它会逐行遍历矩阵 V，每次返回一个 Vector 对象，表示矩阵的一行。
    val toArray:Array[(Vector,Int)] = v.rowIter.zipWithIndex.toArray
    toArray.foreach(println)

    //  todo 注册计算余弦相似度的函数
    //  这里需要先将密集向量转化为数组再使用breeze.linalg.DenseVector()转化为密集向量
    spark.udf.register(
      "cos",
      (v1:DenseVector,v2:DenseVector) => {
        1 - breeze.linalg.functions.cosineDistance(
          breeze.linalg.DenseVector(v1.toArray),breeze.linalg.DenseVector(v2.toArray)
        )
      }
    )

    val sku_info = spark.createDataFrame(
      toArray
    ).toDF("vec", "sku_mapping_id")

    val result=sku_info.crossJoin(sku_info)
      .toDF("vec","sku_mapping_id","vec_2","sku_mapping_id_2")
      .withColumn("cos",expr("cos(vec,vec_2)"))
      .filter(col("sku_mapping_id")!==col("sku_mapping_id_2"))
      .filter(col("sku_mapping_id").isin(nobuy: _*) && col("sku_mapping_id_2").isin(buy: _*))
      .groupBy("sku_mapping_id")
      .agg(
        avg("cos").as("avg_cos")
      )
      .orderBy(desc("avg_cos"))

    println("---------------推荐Top5结果如下---------------------")
    result.limit(5).collect().zipWithIndex.foreach{
      case(r,index) =>{
        val id=r.getAs[Int]("sku_mapping_id")
        val avg=r.getAs[Double]("avg_cos")
        val str=s"相似度top${index+1}(商品id:${id},平均相似度:${avg})"
        println(str)
      }
    }





    spark.close()

  }

}
